package kafka

import (
	log "ac-common-go/glog"
	config_sarama "ac-julink/tool/config_sarama"
	"context"
	"crypto/tls"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/scram"
	"strconv"
	"strings"
	"time"
)

var (
	Writer *kafka.Writer
)

func InitProducers() {
	//initProducers(initSaramaConfig())
	enableSasl := config_sarama.GetConfigBoolValue("enableSasl")
	enableTls := config_sarama.GetConfigBoolValue("enableTls")

	sharedTransport := &kafka.Transport{}
	sharedTransport.DialTimeout = time.Second * time.Duration(config_sarama.GetConfigInt64Value("timeout"))
	sharedTransport.ClientID = config_sarama.GetConfigValue("clientId")

	if enableTls {
		tlsConfig := &tls.Config{
			InsecureSkipVerify: true,
		}
		sharedTransport.TLS = tlsConfig
	}

	if enableSasl {
		username := config_sarama.GetConfigValue("saslUsername")
		password := config_sarama.GetConfigValue("saslPassword")
		saslMechanism := config_sarama.GetConfigValue("saslMechanism")

		var algorithm scram.Algorithm

		if saslMechanism == "SCRAM-SHA-512" {
			algorithm = scram.SHA512
		} else {
			algorithm = scram.SHA256
		}

		mechanism, err := scram.Mechanism(algorithm, username, password)
		if err != nil {
			panic(err)
		}

		sharedTransport.SASL = mechanism
	}

	var seconds int

	if s, err := strconv.Atoi(strings.ReplaceAll(config_sarama.GetConfigValue("flushTimeout"), "s", "")); err != nil {
		log.Error("failed to parse flushTimeout, it must be a '%ds' syntax, use 1 seconds as default.")
		seconds = 1
	} else {
		seconds = s
	}

	batchTimeout := time.Duration(seconds) * time.Second

	Writer = &kafka.Writer{
		Addr:         kafka.TCP(strings.Split(config_sarama.GetConfigValue("brokers"), ",")...),
		Balancer:     &kafka.Hash{},
		RequiredAcks: kafka.RequireNone,
		Async:        true,
		MaxAttempts:  int(config_sarama.GetConfigInt64Value("maxRetries")),
		Transport:    sharedTransport,
		BatchSize:    int(config_sarama.GetConfigInt64Value("batchSize")),
		BatchBytes:   config_sarama.GetConfigInt64Value("maxMessageBytes"),
		BatchTimeout: batchTimeout,
		ReadTimeout:  time.Second * time.Duration(config_sarama.GetConfigInt64Value("readTimeoutSec")),
		WriteTimeout: time.Second * time.Duration(config_sarama.GetConfigInt64Value("writeTimeoutSec")),
	}

	log.Errorf("sarama writer config MaxAttempts: %v", Writer.MaxAttempts)
	log.Errorf("sarama writer config BatchSize: %v", Writer.BatchSize)
	log.Errorf("sarama writer config BatchBytes: %v", Writer.BatchBytes)
	log.Errorf("sarama writer config BatchTimeout: %v", Writer.BatchTimeout)
	log.Errorf("sarama writer config ReadTimeout: %v", Writer.ReadTimeout)
	log.Errorf("sarama writer config WriteTimeout: %v", Writer.WriteTimeout)
}

func ProduceMessageAsync(topic string, msg string, key string) error {
	err := Writer.WriteMessages(context.Background(),
		kafka.Message{
			Topic: topic,
			Key:   []byte(key),
			Value: []byte(msg),
		},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	return nil
}
